package org.collector.util;

import java.util.LinkedList;
import java.util.Queue;

public abstract class ConsumerRunnable<T> extends CanbeClosedRunnable {
	private Queue<T> taskQueue;
	private final static int WAIT_TIME = 5 * 1000;
	public ConsumerRunnable(){
		super();
		taskQueue = new LinkedList<T>();
	}
	
	public boolean addTask(T task){
		if(isClosed())
			return false;
		synchronized(taskQueue){
			boolean rtu = taskQueue.offer(task);
			if(rtu)
				taskQueue.notifyAll();
			return rtu;
		}
	}
	public T pollTask(){
		synchronized(taskQueue){
			return taskQueue.poll();
		}
	}
	private int getTaskQueueSize(){
		synchronized(taskQueue){
			return taskQueue.size();
		}
	}
	public abstract boolean process(T task);
	public abstract void errorHandler(T task);
	@Override
	public void run() {
		// TODO Auto-generated method stub
		while(true){
			T task = null;
			synchronized(taskQueue){
				while(getTaskQueueSize() <= 0){
					try {
						taskQueue.wait(WAIT_TIME);
					} catch (InterruptedException e) {
					}finally{
						if(getTaskQueueSize() <= 0 && isClosed())
							break;
					}
				}
				if(getTaskQueueSize() <= 0 && isClosed())
					break;
				task = pollTask();
			}
			if(task != null){
				if(!process(task))
					errorHandler(task);
			}
		}
	}
	
}
